package pulsar

import (
	"context"
	"fmt"
	"github.com/apache/pulsar-client-go/pulsar"

	"my-zero-tiktok/apps/api/internal/svc"
	"my-zero-tiktok/apps/api/internal/types"

	"github.com/zeromicro/go-zero/core/logx"
)

type PulsarPushLogic struct {
	logx.Logger
	ctx    context.Context
	svcCtx *svc.ServiceContext
}

func NewPulsarPushLogic(ctx context.Context, svcCtx *svc.ServiceContext) *PulsarPushLogic {
	return &PulsarPushLogic{
		Logger: logx.WithContext(ctx),
		ctx:    ctx,
		svcCtx: svcCtx,
	}
}

func (l *PulsarPushLogic) PulsarPush(req *types.PulsarPush) (resp *types.Empty, err error) {
	// 生产者
	errChan := make(chan error, 1)
	go func() {
		var sendErr error
		for i := 0; i < 3; i++ {
			msg := fmt.Sprintf("%s ==> %d", req.Message, i)
			messageID, err := l.svcCtx.PulsarProducer.Send(l.ctx, &pulsar.ProducerMessage{Payload: []byte(msg)})
			if err != nil {
				sendErr = err
				l.Logger.Error("Failed to send message: %v", err)
				break
			}
			l.Logger.Info("Sent message %s with message ID: %s\n", msg, messageID)
		}
		errChan <- sendErr
	}()

	err = <-errChan
	if err != nil {
		return nil, err
	}

	// 消费者
	go func() {
		for {
			msg, err := l.svcCtx.PulsarConsumer.Receive(context.Background())
			if err != nil {
				l.Logger.Errorf("Failed to receive message: %v", err)
			}
			l.Infof("Received message: %s with ID: %s", string(msg.Payload()), msg.ID())
			err = l.svcCtx.PulsarConsumer.Ack(msg)
			if err != nil {
				l.Logger.Errorf("Failed to ack message: %v", err)
			}
		}
	}()

	return &types.Empty{}, nil
}
